package org.example;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.text.ParseException;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

public class CheckLoginFail {

    private   static FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

    public static void main(String[] args) throws Exception {

        //todo: 1、构建流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo: 2、接受socket数据
//        DataStreamSource<String> socketTextStream = env.socketTextStream("localhost", 9999);
        DataStreamSource<String> socketTextStream = env.readTextFile("src/main/resources/log.csv");


        //todo: 3、数据处理分组
        KeyedStream<LoginInfo, Tuple> keyedStream = socketTextStream.map(new MapFunction<String, LoginInfo>() {
            public LoginInfo map(String value) throws Exception {
                String[] users = value.split(",");

                return new LoginInfo(users[0], users[1], users[2], users[3], users[4]);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<LoginInfo>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<LoginInfo>() {
            @Override
            public long extractTimestamp(LoginInfo loginInfo, long l) {
                //指定EventTime对应的字段
                long extractTimestamp = 0;
                try {
                    extractTimestamp = fastDateFormat.parse(loginInfo.eventTime).getTime();
//                    System.out.println(loginInfo.eventTime);
                } catch (ParseException e) {
                    e.printStackTrace();
                }
                return extractTimestamp;
            }//添加水位线
        })).keyBy("userId");

        //todo: 4、定义Parttern
        Pattern<LoginInfo, LoginInfo> pattern = Pattern.<LoginInfo>begin("start").where(new SimpleCondition<LoginInfo>() {
            public boolean filter(LoginInfo value) throws Exception {
                if ("/project/loginFaill".equals(value.url) && value.userId != null) {
                    return true;
                }
                return false;
            }
        }).followedBy("second").where(new IterativeCondition<LoginInfo>() {
            @Override
            public boolean filter(LoginInfo second, Context<LoginInfo> ctx) throws Exception {

                Iterable<LoginInfo> start = ctx.getEventsForPattern("start");
                Iterator<LoginInfo> userLoginIterator = start.iterator();
                while (userLoginIterator.hasNext()) {
                    LoginInfo userLogin = userLoginIterator.next();
                    if ("/project/loginFaill".equals(second.url) && second.userId.equals(userLogin.userId)) {
                        return true;
                    }
                }
                return false;
            }
        }).within(Time.seconds(2)); // 2秒时间约束

        //todo: 5、将Parttern应用到事件流中进行检测，同时指定时间类型
        PatternStream<LoginInfo> patternStream = CEP.pattern(keyedStream, pattern).inEventTime();

        //todo: 6、提取匹配到的数据
        patternStream.select(new PatternSelectFunction<LoginInfo, LoginInfo>() {

            public LoginInfo select(Map<String, List<LoginInfo>> patternMap) throws Exception {

                //Map[String, util.List[(String, UserLoginInfo)]]
                //todo: key就是定义规则的名称：start  second
                //todo: value就是满足对应规则的数据

                List<LoginInfo> start = patternMap.get("start");
                List<LoginInfo> second = patternMap.get("second");

                LoginInfo startData = start.iterator().next();
                LoginInfo secondData = second.iterator().next();

                System.out.println("满足start模式中的数据 ：" + startData);
                System.out.println("满足second模式中的数据：" + secondData);

                return secondData;
            }
        });

        //todo: 7、提交任务
        env.execute("CheckIPChangeWithCepByJava");
    }


    //todo:定义用户登录信息实体对象
    public static class LoginInfo {
        public String ip;
        public String userId;
        public String eventTime;
        public String method;
        public String url;

        @Override
        public String toString() {
            return "LoginInfo{" +
                    "ip='" + ip + '\'' +
                    ", userId='" + userId + '\'' +
                    ", eventTime='" + eventTime + '\'' +
                    ", method='" + method + '\'' +
                    ", url='" + url + '\'' +
                    '}';
        }

        public LoginInfo(String ip, String userId, String eventTime, String method, String url) {
            this.ip = ip;
            this.userId = userId;
            this.eventTime = eventTime;
            this.method = method;
            this.url = url;
        }

        //无参构造必须带上
        public LoginInfo() {
        }

    }
}
